use std::{
    future::Future,
    pin::Pin,
    sync::{
        atomic::{AtomicU32, AtomicU8, Ordering},
        Arc,
    },
    time::Duration,
};

use actix_web::{
    body::{self, BoxBody},
    FromRequest, HttpRequest, HttpResponse, Responder,
};
use dashmap::DashMap;
use serde::Serialize;
use tokio::{
    sync::{mpsc, oneshot, Mutex},
    time::timeout,
};

use crate::{
    error::{new_err, AppErr},
    store::device,
};

use super::codec::RawBuf;

const PING: &'static str = "ping";
const PONG: &'static str = "pong";

const REQ: &'static str = "req";
const RES: &'static str = "res";

const NOTIFY: &'static str = "notify";

const NOTIFY_ACK: &'static str = "notify_ack";
const ACK: &'static str = "ack";

#[derive(Debug, Serialize)]
pub struct DeviceEvent2 {
    seq: u8,

    #[serde(rename = "type")]
    ctype: u8,

    cmd: Option<u8>,

    ec: Option<u8>,

    body: Option<Box<[u8]>>,
}

#[derive(Debug, Serialize)]
pub enum DeviceEvent {
    None,

    Ping(u8),
    Pong(u8),

    Req(u8, u8, Box<[u8]>),
    Res(u8, u8, u8, Box<[u8]>), // cmd ec body

    Notify(u8, Box<[u8]>),

    NotifyAck(u8, u8, Box<[u8]>),
    Ack(u8),
}

pub struct DeviceRemoteInfo {
    event_rx: Mutex<mpsc::Receiver<DeviceEvent>>, // device poll
    event_tx: mpsc::Sender<DeviceEvent>,          // web ctrl
    sequence: AtomicU8,
    resp_map: DashMap<u8, oneshot::Sender<DeviceEvent>>, // device result
    poll_count: AtomicU32,
}

impl DeviceRemoteInfo {

    pub async fn notify(&self, cmd: u8, body: Box<[u8]>) -> Result<(), AppErr> {
        self.event_tx
            .send(DeviceEvent::Notify(cmd, body))
            .await
            .map_err(|_| new_err("send err"))?;
        Ok(())
    }

    pub async fn notify_ack(&self, cmd: u8, body: Box<[u8]>) -> Result<DeviceEvent, AppErr> {
        let seq = self.sequence.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.resp_map.insert(seq, tx);
        self.event_tx
            .send(DeviceEvent::NotifyAck(seq, cmd, body))
            .await
            .map_err(|_| new_err("send err"))?;

        let r = timeout(Duration::from_secs(10), rx)
            .await
            .map_err(|_| new_err("ack timeout"))?
            .map_err(|_| new_err("recv err"))?;

        Ok(r)
    }

    pub async fn recv(&self, duration: Duration) -> Result<DeviceEvent, AppErr> {
        let mut rx = timeout(Duration::from_secs(2), self.event_rx.lock())
            .await
            .map_err(|_| new_err("acquire rx timeout"))?;

        let event = timeout(duration, rx.recv()).await;

        match event {
            Err(_) => return Ok(DeviceEvent::None),
            Ok(Some(v)) => return Ok(v),
            Ok(None) => return Err(new_err("rx recv fail")),
        };
    }
}

pub struct DeviceSession {
    pub id: i64,
}

async fn get_id_by_mac(req: HttpRequest) -> Result<DeviceSession, AppErr> {
    let mac_addr = req
        .headers()
        .get("mac_addr")
        .ok_or(new_err("not mac addr head"))?
        .to_str()
        .map_err(|_| new_err("invalid str"))?;

    let device = device::login(mac_addr).await?;

    Ok(DeviceSession { id: device.id })
}

impl FromRequest for DeviceSession {
    type Error = AppErr;

    type Future = Pin<Box<dyn Future<Output = Result<Self, AppErr>>>>;

    fn from_request(req: &HttpRequest, _payload: &mut actix_web::dev::Payload) -> Self::Future {
        Box::pin(get_id_by_mac(req.clone()))
    }
}

impl Responder for DeviceEvent {
    type Body = BoxBody;

    fn respond_to(self, _req: &HttpRequest) -> HttpResponse {
        match self {
            Self::None => HttpResponse::Ok().insert_header(("type", "none")).finish(),

            Self::Ping(seq) => HttpResponse::Ok()
                .insert_header(("type", PING))
                .insert_header(("seq", format!("{}", seq)))
                .finish(),

            Self::Pong(seq) => HttpResponse::Ok()
                .insert_header(("type", PONG))
                .insert_header(("seq", format!("{}", seq)))
                .finish(),

            Self::Ack(seq) => HttpResponse::Ok()
                .insert_header(("type", ACK))
                .insert_header(("seq", format!("{}", seq)))
                .finish(),

            Self::Notify(cmd, body) => HttpResponse::Ok()
                .insert_header(("type", NOTIFY))
                .insert_header(("cmd", format!("{}", cmd)))
                .body(RawBuf(body)),

            Self::NotifyAck(seq, cmd, body) => HttpResponse::Ok()
                .insert_header(("type", NOTIFY_ACK))
                .insert_header(("cmd", format!("{}", cmd)))
                .insert_header(("seq", format!("{}", seq)))
                .body(RawBuf(body)),

            Self::Req(seq, cmd, body) => HttpResponse::Ok()
                .insert_header(("type", REQ))
                .insert_header(("cmd", format!("{}", cmd)))
                .insert_header(("seq", format!("{}", seq)))
                .body(RawBuf(body)),

            Self::Res(seq, cmd, ec, body) => HttpResponse::Ok()
                .insert_header(("type", REQ))
                .insert_header(("seq", format!("{}", seq)))
                .insert_header(("cmd", format!("{}", cmd)))
                .insert_header(("ec", format!("{}", ec)))
                .body(RawBuf(body)),
        }
    }
}
